package demo;


import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;


import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

public class SparkSortByDemo {
    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf().setAppName("spark demo").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        List<Person> person = new ArrayList<>();


        person.add(new Person("DD", 18));
        person.add(new Person("CC", 19));
        person.add(new Person("BB", 120));
        person.add(new Person("FF", 18));

        JavaRDD<Person> rdd = sc.parallelize(person);
        Function<Person, Person> function = (new Function<Person, Person>() {
            @Override
            public Person call(Person person) throws Exception {
                return person;
            }
        });

        JavaRDD<Person> newRDD = rdd.sortBy(function, true, rdd.getNumPartitions());

        List<Person> res = newRDD.collect();
        for (Person i :res){
            System.out.println(i);
        }
        
    }
}

class Person implements Serializable, Comparable {
    private String name;
    private int age;

    public Person() {
    }

    public Person(String name, int age) {
        this.name = name;
        this.age = age;
    }

    public String getName() {
        return name;
    }

    public int getAge() {
        return age;
    }

    public void setName(String name) {
        this.name = name;
    }

    public void setAge(int age) {
        this.age = age;
    }

    @Override
    public String toString() {
        return "Person{" +
                "name='" + name + '\'' +
                ", age=" + age +
                '}';
    }

    @Override
    public int compareTo(Object o) {
        Person p = (Person)o;
        if (this.getAge() > p.getAge()) {
            return 1;
        } else if (this.getAge() == p.getAge()) {
            return 0;
        } else {
            return -1;
        }
    }
}
